/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package aiaudio.processing.ratings.inverse;

import aiaudio.processing.base.mapreduce.UserValuePair;
import aiaudio.database.tables.RatingsListTable;
import java.io.IOException;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.mapreduce.Reducer;

/**
 *
 * @author nastya
 */
class ByUserMapper extends TableMapper<ImmutableBytesWritable, UserValuePair> {
    private int numRecords = 0;

    @Override
    protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
        try {
            byte[] artistKey = RatingsListTable.getArtistAsBytes(value);
            byte[] userKey = RatingsListTable.getUserAsBytes(value);
            ImmutableBytesWritable immutableArtistKey = new ImmutableBytesWritable(artistKey);
            int listenCount = RatingsListTable.getListenCount(value);
            context.write(immutableArtistKey, new UserValuePair(listenCount, userKey));
        } catch (InterruptedException e) {
            throw new IOException(e);
        }
        numRecords++;
        if ((numRecords % 1000) == 0) {
            context.setStatus("mapper processed " + numRecords + " records so far");
        }
    }
    
}
